package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.UserLoginBean;
import com.atguigu.edu.realtime.util.ClickhouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsUserLoginWindow {
    public static void main(String[] args) throws Exception {

        //0 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //1数据源：dwd_traffic_page_log 用户的页面浏览
         //1  获取数据流
        String sourceTopic="dwd_traffic_page_log";
        String groupId="dws_user_login_window";
        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(JSON::parseObject);

        //2筛选出用户的登录行为
        //   两种登录方式：1  启动时自动登录   会话首页+uid不为空
        //              2  半路手动登录       找到上一个页面为login页
        SingleOutputStreamOperator<JSONObject> loginJsonObj = jsonObjStream.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                JSONObject commonObj = jsonObject.getJSONObject("common");
                JSONObject pageObj = jsonObject.getJSONObject("page");
                if ((pageObj.getString("last_page_id") == null || pageObj.getString("last_page_id").length() == 0)
                        && (commonObj.getString("uid") != null && commonObj.getString("uid").length() > 0)) {
                    //1  启动时自动登录   会话首页+uid不为空
                    return true;
                } else if (pageObj.getString("last_page_id") != null && pageObj.getString("last_page_id").equals("login")
                        && (commonObj.getString("uid") != null && commonObj.getString("uid").length() > 0)) {
                    // 2  半路手动登录       找到上一个页面为login页 且uid不为null
                    return true;
                }
                return false;
            }
        });
      //  loginJsonObj.print();
        //3筛选：利用状态筛选出当日首次登录
        //        回流的用户的判断
        KeyedStream<JSONObject, String> uidKeyedStream = loginJsonObj.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("uid"));
        SingleOutputStreamOperator<JSONObject> uniqueUserStream = uidKeyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            //状态： 用户最后一次登录日期
            ValueStateDescriptor<String> userLastLoginDtStateDesr = new ValueStateDescriptor<String>("user_last_login_dt", String.class);
            ValueState<String> userLastLoginDtState = null;
            @Override
            public void open(Configuration parameters) throws Exception {

                userLastLoginDtState = getRuntimeContext().getState(userLastLoginDtStateDesr);
            }
            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                //1  非独立用户过滤掉 // 状态为空  或者 不为空 但是不是今天  -->独立登录用户 保留
                //  2  状态不为空  距离现在超过7天  -->回流用户   增加一个表示 is_back
                Long ts = jsonObject.getLong("ts");
                String curDate = DateFormatUtil.toDate(ts);
                if (userLastLoginDtState.value() == null || userLastLoginDtState.value().length() == 0) {//状态为空  -->独立登录用户 保留

                    out.collect(jsonObject);
                    userLastLoginDtState.update(curDate);
                } else if (!userLastLoginDtState.value().equals(curDate)) {  //不为空 但是不是今天  -->独立登录用户 保留
                    //距离现在超过7天
                    Long lastLoginTs = DateFormatUtil.toTs(userLastLoginDtState.value());
                    long intervalDays = (ts - lastLoginTs) / 1000 / 60 / 60 / 24;
                    if (intervalDays >= 7) {
                        jsonObject.put("is_back", "1");
                    }
                    out.collect(jsonObject);
                    userLastLoginDtState.update(curDate);
                }
            }
        });
        uniqueUserStream.print();
        //4调整结构
        SingleOutputStreamOperator<UserLoginBean> loginBeanStream = uniqueUserStream.map(new MapFunction<JSONObject, UserLoginBean>() {
            @Override
            public UserLoginBean map(JSONObject loginObj) throws Exception {
                String isBack = loginObj.getString("is_back");
                Long backCt = 0L;
                if ("1".equals(isBack)) {
                    backCt = 1L;
                }

                return new UserLoginBean("", "", backCt, 1L, loginObj.getLong("ts"));
            }
        });

        //5开窗 聚合
        //5.1 水位线
        SingleOutputStreamOperator<UserLoginBean> userLoginBeanWmStream = loginBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserLoginBean>forBoundedOutOfOrderness(Duration.ofMillis(1000)).withIdleness(Duration.ofMillis(10000))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserLoginBean>() {
                    @Override
                    public long extractTimestamp(UserLoginBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //5.2开窗聚合
        SingleOutputStreamOperator<UserLoginBean> reduceStream = userLoginBeanWmStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean bean1, UserLoginBean bean2) throws Exception {
                        bean1.setBackCt(bean1.getBackCt() + bean2.getBackCt());
                        bean1.setUuCt(bean1.getUuCt() + bean2.getUuCt());
                        return bean1;
                    }
                }, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserLoginBean> beanItr, Collector<UserLoginBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        for (UserLoginBean userLoginBean : beanItr) {
                            userLoginBean.setStt(stt);
                            userLoginBean.setEdt(edt);
                            userLoginBean.setTs(System.currentTimeMillis());
                            out.collect(userLoginBean);
                        }
                    }
                });

        //6  写入clickhouse
        reduceStream.addSink(ClickhouseUtil.getSink("insert into dws_user_user_login_window values(?,?,?,?,?)"));
        env.execute();
    }
}
